Summary

Implemented connection health checks and scoped database sessions to the message processing lifecycle to prevent stale connections and resource leaks.

This fixes an issue where the GCS importer would get broken connections to the postgres container.

Technical Details:

• Connection Resilience: Added pool_pre_ping=True to the SQLAlchemy engine configuration in src/datastores/sql/database.py. This ensures the engine tests the validity of a connection before using it, automatically reconnecting if the database server has closed the socket.
• Session Lifecycle Management: Refactored the GCP importer to use a context manager for database sessions within the Pub/Sub callback. By moving database.SessionLocal() inside the callback function, the system now creates and closes a fresh session for every message processed.
• Resource Cleanup: Removed the long-lived database session from the main execution loop. This change eliminates the risk of connection timeouts in long-running subscriber processes and removes the need for manual session closing in the finally block.